package com.pk.flink.scenario01.source;

import com.alibaba.fastjson.JSONArray;
import com.pk.flink.scenario01.bean.NotFoundElement;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.util.FileUtils;

import java.io.File;

public class NotFoundElementSource extends RichParallelSourceFunction<NotFoundElement> {
    private boolean isRunning = true;
    @Override
    public void run(SourceContext<NotFoundElement> ctx) throws Exception {
        String fileContent = FileUtils.readFile(new File("D:\\sources\\pk-flink\\data\\not_found_element_results.json"), "utf-8");
        JSONArray jsonArray = JSONArray.parseArray(fileContent);
        for (int i = 0; i < jsonArray.size(); i++) {
            NotFoundElement notFoundElement = jsonArray.getObject(i, NotFoundElement.class);
            ctx.collect(notFoundElement);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
